package com.wang.lottery.application.worker;

import cn.bugstack.middleware.db.router.strategy.IDBRouterStrategy;
import com.alibaba.fastjson.JSON;
import com.wang.lottery.application.mp.producer.KafkaProducer;
import com.wang.lottery.common.Constants;
import com.wang.lottery.common.Result;
import com.wang.lottery.domain.activity.model.po.Activity;
import com.wang.lottery.domain.activity.model.vo.InvoiceVO;
import com.wang.lottery.domain.activity.service.deploy.IActivityDeploy;
import com.wang.lottery.domain.activity.service.partake.IActivityPartake;
import com.wang.lottery.domain.activity.service.stateflow.IStateHandler;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * @author wjh
 * @create 2023-11-15 14:07
 * 抽奖业务，任务配置
 */
@Component
@Slf4j
public class LotteryXxlJob {
    
    @Resource
    private IActivityDeploy  activityDeploy;
    @Resource
    private IStateHandler stateHandler;
    @Resource
    private IActivityPartake activityPartake;
    @Resource
    private IDBRouterStrategy dbRouter;
    @Resource
    private KafkaProducer kafkaProducer;
    /**
     * 1.扫描活动状态，将状态为审核通过（state=4）的活动，改为进行中
     *      1.1：即使将活动提前开启，用户也是参与不了活动的，因为在用户参与的时候除了会判断活动状态还要判断当前时间是否是活动时间，
     *      所以这里不必担心用户可以提前参与活动
     * 2.将已结束的活动，但状态还是进行中的活动改为已结束
     * 
     */
    @XxlJob("lotteryActivityStateJobHandler")
    public void lotteryActivityStateJobHandler() throws Exception{
        log.info("扫描活动状态 Begin");
        List<Activity> activityList = activityDeploy.scanToDoActivityList(0L);
        if (activityList.isEmpty()) {
            log.info("扫描活动状态 End，没有需要处理的活动");
            return;
        }
        while (!activityList.isEmpty()) {
            for (Activity activity : activityList) {
                switch (activity.getState()) {
                    case 4:
                        Result doing = stateHandler.doing(activity.getActivityId(), Constants.ActivityState.PASS);
                        //打印日志
                        log.info("扫描活动状态为活动中 结果：{} activityId：{} activityName：{} creator：{}", JSON.toJSONString(doing), activity.getActivityId(), activity.getActivityName(), activity.getCreator());
                        break;
                    case 5:
                        if (activity.getEndDateTime().before(new Date())) {
                            Result close = stateHandler.close(activity.getActivityId(), Constants.ActivityState.DOING);
                            log.info("扫描活动状态为活动结束 结果：{} activityId：{} activityName：{} creator：{}", JSON.toJSONString(close), activity.getActivityId(), activity.getActivityName(), activity.getCreator());
                        }
                        break;
                    default:
                        break;
                }
            }
            // 获取集合中最后一条记录，继续扫描后面10条记录
            activityList = activityDeploy.scanToDoActivityList(activityList.get(activityList.size() - 1).getId());
        }
        log.info("扫描活动状态 End");
    }
    
    @XxlJob("lotteryOrderMQStateJobHandler")
    public void lotteryOrderMQStateJobHandler() throws Exception{
        //获取参数
        String param = XxlJobHelper.getJobParam();
        // 验证参数
        if (param == null) {
            log.info("扫描用户抽奖奖品发放MQ状态[Table = 2*4] 错误 params is null");
            return;
        }
        // 获取分布式任务配置参数信息 参数配置格式：1,2,3 也可以是指定扫描一个，也可以配置多个库，按照部署的任务集群进行数量配置，均摊分别扫描效率更高
        String[] dbNumbers = param.split(",");
        if (dbNumbers.length < 1) {
            log.info("扫描用户抽奖奖品发放MQ状态[Table = 2*4] 错误 params is null");
            return;
        }
        //分库数量
        int dbCount = dbRouter.dbCount();
        //分表数量
        int tbCount = dbRouter.tbCount();
        // 循环获取指定扫描库
        for (String dbNumber : dbNumbers) {
            // 判断配置指定扫描库数，是否存在
            if (Integer.parseInt(dbNumber) > dbCount) {
                log.info("扫描用户抽奖奖品发放MQ状态[Table = 2*4] 错误 dbCount > dbRouter.dbCount()");
                continue;
            }
            //循环扫描表
            for (int i = 0; i < tbCount; i++) {
                // 扫描库表数据
                List<InvoiceVO> invoiceVOList = activityPartake.scanInvoiceMqState(Integer.parseInt(dbNumber), i);
                if (invoiceVOList.isEmpty()) {
                    continue;
                }
                // 补偿 MQ 消息
                for (InvoiceVO invoiceVO : invoiceVOList) {
                    ListenableFuture<SendResult<String, Object>> resultListenableFuture = kafkaProducer.sendLotteryInvoice(invoiceVO);
                    resultListenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            log.error("发送MQ失败，原因：{}", ex.getMessage());
                            // MQ 消息发送失败，更新数据库表 user_strategy_export.mq_state = 2 【等待定时任务扫码补偿MQ消息】
                            activityPartake.updateInvoiceMqState(invoiceVO.getuId(),invoiceVO.getOrderId(),2);
                        }
                        
                        @Override
                        public void onSuccess(SendResult<String, Object> result) {
                            log.info("发送MQ成功，结果：{}", result.toString());
                            // MQ 消息发送完成，更新数据库表 user_strategy_export.mq_state = 1
                            activityPartake.updateInvoiceMqState(invoiceVO.getuId(),invoiceVO.getOrderId(),1);
                        }
                    });
                }
            }
        }
        log.info("扫描用户抽奖奖品发放MQ状态[Table = 2*4] 完成 param：{}", JSON.toJSONString(dbNumbers));
    }
}
